package com.xxl.job.executor.kafka;

import com.xxl.job.core.log.XxlJobLogger;
import com.xxl.job.executor.common.KafkaTopicConstant;
import com.xxl.job.executor.model.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * kafka生产端
 */
@Service
public class KafkaProduceService {

    private ExecutorService executor = Executors.newFixedThreadPool(50);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendHkAlarmMessage(CurrentInfoHkModel message){
        sendMessage(KafkaTopicConstant.HK_ALARM_TOPIC_MESSAGE,message.toJson());
    }

    public void sendBLMessage(CurrentInfoHkModel message){
        sendMessage(KafkaTopicConstant.BL_TOPIC_MESSAGE,message.toJson());
    }

    public void sendRqAlarmMessage(CurrentInfoRqModel message){
        sendMessage(KafkaTopicConstant.RQ_ALARM_TOPIC_MESSAGE,message.toJson());
    }

    public void sendXfAlarmMessage(CurrentInfoXfModel message){
        sendMessage(KafkaTopicConstant.XF_ALARM_TOPIC_MESSAGE,message.toJson());
    }

    public void sendDhAlarmMessage(CurrentInfoDhModel message){
        sendMessage(KafkaTopicConstant.DH_ALARM_TOPIC_MESSAGE,message.toJson());
    }


   /*门禁kafka*/
    public void sendMjAlarmMessage(CurrentInfoMjModel message){
        sendMessage(KafkaTopicConstant.MJ_ALARM_TOPIC_MESSAGE,message.toJson());
    }



    public void sendBkAlarmMessage(JsonBkInterfaceModel message){
        sendMessage(KafkaTopicConstant.BK_ALARM_TOPIC_MESSAGE,message.toJson());
    }


    public void sendBkDevMessage(JsonBkInterfaceModel message){
        sendMessage(KafkaTopicConstant.BK_DEV_TOPIC_MESSAGE,message.toJson());
    }

    /**
     * 多线程发送消息到 Kafka
     */
    public void sendMessage(String topic, String message) {
        executor.execute(() -> asyncSendMessage(topic, message));
    }

    /**
     * 同步发送消息
     */
    public void syncSendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }

    /**
     * 异步发送消息
     */
    public void asyncSendMessage(String topic, String message) {
        // the KafkaTemplate provides asynchronous send methods returning a Future
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);

        // register a callback with the listener to receive the result of the send asynchronously
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

            @Override
            public void onSuccess(SendResult<String, String> result) {
                //  XxlJobLogger.log("SUCCESS: send message='{}' to topic='{}' with offset={}", message, topic, result.getRecordMetadata().offset());
            }

            @Override
            public void onFailure(Throwable ex) {
                XxlJobLogger.log("ERROR: unable to send message='{}' to topic='{}'", message, topic, ex);
            }
        });
    }

}
